package com.neuedu.brazil;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.icepear.echarts.Line;
import org.icepear.echarts.render.Engine;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.*;


/**
 * Hadoop自定义数据类型
 *
 * @author DEll
 */

public class Manager {
    public static Map<String, String> PATHS = new HashMap<String, String>();
    private static final Scanner scanner = new Scanner(System.in);

    static {
        PATHS.put("step1_input", "/brazil_weather");
        PATHS.put("step1_output", "/step1_output");

        PATHS.put("step2_input", "/step1_output");
        PATHS.put("step2_output", "/step2_output");

        PATHS.put("step3_input", "/step1_output");
        PATHS.put("step3_output", "/step3_output");

        PATHS.put("step4_input", "/step1_output");
        PATHS.put("step4_output", "/step4_output");

        PATHS.put("step5_input", "/step1_output");
        PATHS.put("step5_output", "/step5_output");

    }

    /**
     * 输出满足查询日期的天气数据
     *
     * @param path
     * @return
     */
    private static List<WeatherWritable> read2Weather(String path) {
        List<WeatherWritable> weathers = new ArrayList<WeatherWritable>();
        try {
            Path dst = new Path(path);
            for (FileStatus f : HadoopUtils.getFs().listStatus(dst)) {
                if (f.isFile()) {
                    FSDataInputStream inputStream = HadoopUtils.getFs().open(f.getPath());
                    BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
                    String line = reader.readLine();
                    while (line != null) {
                        String[] items = line.split("\t");
                        String code = items[0];
                        String date = items[1];
                        double precipitation = Double.parseDouble(items[2]);
                        double maxTemperature = Double.parseDouble(items[3]);
                        double minTemperature = Double.parseDouble(items[4]);
                        double avgTemperature = Double.parseDouble(items[5]);
                        WeatherWritable w = new WeatherWritable(code, date, precipitation, maxTemperature, minTemperature, avgTemperature);
                        weathers.add(w);
                        line = reader.readLine();
                    }
                    reader.close();
                    inputStream.close();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return weathers;
    }

    private static void show(List<WeatherWritable> weathers) {
        for (WeatherWritable w : weathers) {
            System.out.println(w);
        }
    }

    public static void printMenu() {
        System.out.println("*******************************");
        System.out.println("1.数据清洗与数据导入");
        System.out.println("2.查询指定日期的天气");
        System.out.println("3.汇总统计每年的天气");
        System.out.println("4.预测指定日期的天气");
        System.out.println("5.预测未来15天的天气");
        System.out.println("6.每年的天气温度的数据可视化");
        System.out.println("7.");
        System.out.println("8.");
        System.out.println("0.退出");
        System.out.println("*****************************");
        System.out.println("请输入你的选项（0-8):");
    }

    public static void selectMenu() {
        int choice = -1;
        boolean isExited = false;
        do {
            printMenu();
            choice = scanner.nextByte();
            switch (choice) {
                case 1:
                    dataImport();
                    break;
                case 2:
                    searchWeather();
                    break;
                case 3:
                    groupByYear();
                    break;
                case 4:
                    predictWeather4Date();
                    break;
                case 5:
                    predictWeatherOff15Days();
                    break;
                case 6:
                    echartsOfYear();
                    break;
                case 7:
                    break;
                case 8:
                    break;
                case 0:
                    isExited = true;
                    break;
                default:
                    System.out.println("输入有误，请重新输入");
            }
        } while (!isExited);
        System.out.println("谢谢使用");
    }

    private static void predictWeatherOff15Days() {
        //多步算法：预测未来15天的天气
        Step5.run(PATHS.get("step5_input"), PATHS.get("step5_output"));
        List<WeatherWritable> weathers = read2Weather(PATHS.get("step5_output"));
        show(weathers);
    }

    private static void predictWeather4Date() {
        //多步算法4：预测指定日期的天气
        System.out.println("请输入要预测的日期（dd/MM)");
        String date = scanner.next();
        Step4.run(PATHS.get("step4_input"), PATHS.get("step4_output"), date);
        List<WeatherWritable> weathers = read2Weather(PATHS.get("step4_output"));
        show(weathers);
    }

    private static void echartsOfYear() {
        List<WeatherWritable> weathers = read2Weather(PATHS.get("step3_output"));
        int length = weathers.size();
        String[] x = new String[length];
        for (int i = 0; i < length; i++) {
            x[i] = weathers.get(i).getDate();
        }
        Number[] y1 = new Number[length];
        Number[] y2 = new Number[length];
        Number[] y3 = new Number[length];
        for (int i = 0; i < length; i++) {
            y1[i] = weathers.get(i).getMaxTemperature();
            y2[i] = weathers.get(i).getMinTemperature();
            y3[i] = weathers.get(i).getAvgTemperature();
        }
        Line line = new Line()
                .addXAxis("年份", x)
                .addYAxis()
                .addSeries("最高温度", y1)
                .addSeries("最低温度", y2)
                .addSeries("平均温度", y3)
                .setTitle("巴西利亚每年的温度系统")
                .setLegend();
        Engine engine = new Engine();
        String fileName = "weatherOfYear.html";
        engine.render(fileName, line);
        try {
            Runtime.getRuntime().exec("explorer " + fileName);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static void groupByYear() {
        //多步算法3：汇总统计每年的天气
        Step3.run(PATHS.get("step3_input"), PATHS.get("step3_output"));
        List<WeatherWritable> weathers = read2Weather(PATHS.get("step3_output"));
        show(weathers);
    }

    private static void searchWeather() {
        //多步算法2：查询指定日期的天气
        System.out.println("请输入要查询的日期：");
        String date = scanner.next();
        Step2.run(PATHS.get("step2_input"), PATHS.get("step2_output"), date);
        List<WeatherWritable> weathers = read2Weather(PATHS.get("step2_output"));
        show(weathers);
    }

    private static void dataImport() {
        //多步算法1：数据清洗与数据导入
        Step1.run(PATHS.get("step1_input"), PATHS.get("step1_output"));
    }

    public static void main(String[] args) {
        selectMenu();
        //多步算法1：数据清洗与数据导入
        //Step1.run(PATHS.get("step1_input"),PATHS.get("step1_output"));
        //多步算法2：查询指定日期的天气
        //System.out.println("请输入要查询的日期：");
        //String date = scanner.next();
        //Step2.run(PATHS.get("step2_input"),PATHS.get("step2_output"),date);
        //List<WeatherWritable> weathers = read2Weather(PATHS.get("step2_output"));
        //show(weathers);
        //多步算法3：汇总统计每年的天气
        //Step3.run(PATHS.get("step3_input"),PATHS.get("step3_output"));
        //List<WeatherWritable> weathers = read2Weather(PATHS.get("step3_output"));
        //show(weathers);
        //多步算法4：预测指定日期的天气
        //System.out.println("请输入要预测的日期（dd/MM)");
        //String date = scanner.next();
        //Step4.run(PATHS.get("step4_input"), PATHS.get("step4_output"),date );
        //List<WeatherWritable> weathers = read2Weather(PATHS.get("step4_output"));
        //show(weathers);
        //多步算法：预测未来15天的天气
        //Step5.run(PATHS.get("step5_input"),PATHS.get("step5_output"));
        //List<WeatherWritable> weathers = read2Weather(PATHS.get("step5_output"));
        //show(weathers);

    }
}
